package sg.edu.nus.iss.stockex;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedList;


import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;

/**
 * MovingAverageReducerWithoutRunningWindow
 * 
 */
public class MovingAverageReducerWithoutRunningWindow extends Reducer<CodeDateKey, TimestampPriceDataPoint, Text, Text> {

	static long DAY_IN_MS = 24 * 60 * 60 * 1000;

	@Override
	public void reduce(CodeDateKey key, Iterable<TimestampPriceDataPoint> values,
			Context context) throws IOException, InterruptedException {
		
				// ---------- compute the moving average here -----------
				
				double sumOpen = 0;
				double sumClose = 0;
				double sumHigh = 0;
				double sumLow = 0;
				int count = 0;
				TimestampPriceDataPoint lastDataPoint = null;
				
				for (TimestampPriceDataPoint dataPoint: values){
					count++;
					sumClose += dataPoint.getClosePrice();
					sumOpen += dataPoint.getOpenPrice();
					sumHigh += dataPoint.getHighPrice();
					sumLow += dataPoint.getLowPrice();
					lastDataPoint = dataPoint;
				}
				
				double movingAverageClose = sumClose/count;
				double movingAverageOpen = sumOpen/count;
				double movingAverageHigh = sumHigh/count;
				double movingAverageLow = sumLow/count;
				
				Text outKey = new Text();
				Text outVal = new Text();
				
				/*
				outKey.set("Code: " + key.getCode() + ", Date: "
						+ lastDataPoint.getDate());
				outVal.set("Moving Average: " + movingAverage);
				 */
				
				outKey.set(key.getCode() + "\t" + lastDataPoint.getDate());
				outVal.set(movingAverageHigh + "\t" + movingAverageLow + "\t" + movingAverageClose + "\t" + movingAverageOpen );
				
				context.write(outKey, outVal);

	} // end reduce

}
